LangGraph Trimming 集成方案

一、模块概述

属性	说明
模块名称	Trimming（消息裁剪）+ Summarization（对话摘要）
优先级	🟡 P2（中）
预估工时	0.5-1 天
依赖项	langchain-core (trim_messages)

为什么需要

当前实现没有处理长对话超出上下文窗口的问题：

• 长对话可能超出模型上下文窗口限制
• 过多历史消息会增加 API 成本
• 旧消息的相关性降低，但关键信息需要保留

---

二、方案对比

2.1 Trimming vs Summarization

特性	Trimming（裁剪）	Summarization（摘要）
原理	直接删除旧消息	用摘要替代旧消息
信息保留	完全丢失	保留关键信息
实现复杂度	低	中
API 调用	无需额外调用	需要调用 LLM 生成摘要
适用场景	简单对话	需要保留上下文的长对话
推荐	默认方案	高级方案

2.2 推荐策略

┌───────────────────────────────────────────────────────────────┐
│                    记忆管理策略选择                            │
├───────────────────────────────────────────────────────────────┤
│                                                               │
│   对话长度 < 阈值                                              │
│   ┌─────────────────┐                                        │
│   │   不做处理       │ ──────────────────────────────────────► │
│   │   直接使用       │                                        │
│   └─────────────────┘                                        │
│                                                               │
│   对话长度 > 阈值                                              │
│   ┌─────────────────┐                                        │
│   │   Trimming      │ ──► 裁剪到最近 N 条消息                  │
│   │   (快速方案)     │                                        │
│   └─────────────────┘                                        │
│           │                                                   │
│           ▼                                                   │
│   ┌─────────────────┐                                        │
│   │  Summarization  │ ──► 生成摘要 + 保留最近消息              │
│   │  (高级方案)      │                                        │
│   └─────────────────┘                                        │
│                                                               │
└───────────────────────────────────────────────────────────────┘

---

三、代码实现

3.1 Trimming 实现

创建文件: services/message_trimmer.py

"""消息裁剪服务

当对话过长时，裁剪旧消息以适应上下文窗口。
"""
import logging
from typing import List, Optional, Callable
from langchain_core.messages import BaseMessage, trim_messages, HumanMessage, AIMessage

logger = logging.getLogger(__name__)


# Token 计数器（近似值，实际应使用 tiktoken）
def count_tokens_approximately(message: BaseMessage) -> int:
    """近似计算消息的 token 数量

    中文约 1.5 字/token，英文约 4 字符/token
    这里使用简单估算：字符数 / 2
    """
    content = getattr(message, 'content', '') or ''
    return max(1, len(str(content)) // 2)


class MessageTrimmer:
    """消息裁剪器"""

    def __init__(
        self,
        max_tokens: int = 4000,
        strategy: str = "last",
        start_on: str = "human",
        end_on: tuple = ("human", "tool"),
        token_counter: Optional[Callable] = None
    ):
        """
        初始化裁剪器

        Args:
            max_tokens: 最大 token 数量
            strategy: 裁剪策略，"last" 保留最近的
            start_on: 裁剪后的第一条消息类型
            end_on: 裁剪后的最后一条消息类型
            token_counter: token 计数函数
        """
        self.max_tokens = max_tokens
        self.strategy = strategy
        self.start_on = start_on
        self.end_on = end_on
        self.token_counter = token_counter or count_tokens_approximately

    def trim(self, messages: List[BaseMessage]) -> List[BaseMessage]:
        """
        裁剪消息列表

        Args:
            messages: 原始消息列表

        Returns:
            裁剪后的消息列表
        """
        if not messages:
            return messages

        # 计算当前 token 数
        total_tokens = sum(self.token_counter(m) for m in messages)

        if total_tokens <= self.max_tokens:
            # 不需要裁剪
            return messages

        logger.info(f"消息总 token 数 {total_tokens} 超过阈值 {self.max_tokens}，开始裁剪")

        try:
            trimmed = trim_messages(
                messages,
                max_tokens=self.max_tokens,
                strategy=self.strategy,
                token_counter=self.token_counter,
                start_on=self.start_on,
                end_on=self.end_on,
            )

            new_tokens = sum(self.token_counter(m) for m in trimmed)
            logger.info(f"裁剪完成: {len(messages)} -> {len(trimmed)} 条消息, "
                       f"{total_tokens} -> {new_tokens} tokens")

            return trimmed

        except Exception as e:
            logger.error(f"裁剪消息失败: {e}")
            # 裁剪失败时，保留最后 N 条消息
            return messages[-10:]  # 保留最后 10 条

    def get_stats(self, messages: List[BaseMessage]) -> dict:
        """获取消息统计信息"""
        if not messages:
            return {"count": 0, "tokens": 0}

        total_tokens = sum(self.token_counter(m) for m in messages)
        return {
            "count": len(messages),
            "tokens": total_tokens,
            "human_messages": sum(1 for m in messages if isinstance(m, HumanMessage)),
            "ai_messages": sum(1 for m in messages if isinstance(m, AIMessage)),
        }


# 全局默认裁剪器
_default_trimmer: Optional[MessageTrimmer] = None


def get_trimmer(
    max_tokens: int = 4000,
    strategy: str = "last"
) -> MessageTrimmer:
    """获取消息裁剪器"""
    global _default_trimmer
    if _default_trimmer is None:
        _default_trimmer = MessageTrimmer(max_tokens=max_tokens, strategy=strategy)
    return _default_trimmer

3.2 Summarization 实现

创建文件: services/message_summarizer.py

"""对话摘要服务

当对话过长时，生成摘要替代旧消息。
"""
import logging
from typing import List, Optional, TypedDict
from langchain_core.messages import BaseMessage, HumanMessage, AIMessage, SystemMessage
from langchain_openai import ChatOpenAI
import os

logger = logging.getLogger(__name__)


class SummaryState(TypedDict):
    """摘要状态"""
    messages: List[BaseMessage]
    summary: Optional[str]


class MessageSummarizer:
    """对话摘要器"""

    SUMMARY_PROMPT = """请将以下对话历史总结为简洁的摘要，保留关键信息：

## 对话历史
{conversation}

## 摘要要求
1. 保留关键的用户需求和偏好
2. 保留重要的决策和结论
3. 使用简洁的语言
4. 不超过 200 字

## 摘要
"""

    def __init__(
        self,
        max_messages: int = 20,
        keep_recent: int = 4,
        model: Optional[str] = None
    ):
        """
        初始化摘要器

        Args:
            max_messages: 触发摘要的消息数量阈值
            keep_recent: 保留最近的消息数量
            model: 用于生成摘要的模型
        """
        self.max_messages = max_messages
        self.keep_recent = keep_recent
        self.llm = ChatOpenAI(
            model=model or os.getenv("OPENROUTER_MODEL", "openai/gpt-4o-mini"),
            api_key=os.getenv("OPENROUTER_API_KEY"),
            base_url=os.getenv("OPENROUTER_BASE_URL", "https://openrouter.ai/api/v1"),
            temperature=0.3
        )

    def should_summarize(self, messages: List[BaseMessage]) -> bool:
        """判断是否需要生成摘要"""
        return len(messages) > self.max_messages

    def summarize(self, messages: List[BaseMessage], existing_summary: Optional[str] = None) -> str:
        """
        生成对话摘要

        Args:
            messages: 消息列表
            existing_summary: 已有的摘要（如果有）

        Returns:
            新的摘要文本
        """
        if not messages:
            return existing_summary or ""

        # 格式化对话
        conversation = self._format_conversation(messages)

        # 构建提示
        if existing_summary:
            prompt = f"""这是之前的对话摘要：
{existing_summary}

请结合新消息扩展摘要：

新消息：
{conversation}
"""
        else:
            prompt = self.SUMMARY_PROMPT.format(conversation=conversation)

        try:
            response = self.llm.invoke([
                SystemMessage(content="你是一个专业的对话摘要助手。"),
                HumanMessage(content=prompt)
            ])

            new_summary = response.content
            logger.info(f"生成摘要成功，长度: {len(new_summary)}")

            return new_summary

        except Exception as e:
            logger.error(f"生成摘要失败: {e}")
            return existing_summary or "摘要生成失败"

    def _format_conversation(self, messages: List[BaseMessage]) -> str:
        """格式化对话为文本"""
        lines = []
        for msg in messages:
            role = "用户" if isinstance(msg, HumanMessage) else "AI" if isinstance(msg, AIMessage) else "系统"
            content = str(msg.content)[:200]  # 限制单条消息长度
            if len(str(msg.content)) > 200:
                content += "..."
            lines.append(f"[{role}]: {content}")
        return "\n".join(lines)

    def process_with_summary(
        self,
        messages: List[BaseMessage],
        existing_summary: Optional[str] = None
    ) -> tuple:
        """
        处理消息，生成摘要并裁剪

        Args:
            messages: 原始消息列表
            existing_summary: 已有的摘要

        Returns:
            (处理后的消息, 新摘要)
        """
        if not self.should_summarize(messages):
            return messages, existing_summary

        # 需要摘要的消息
        messages_to_summarize = messages[:-self.keep_recent]
        recent_messages = messages[-self.keep_recent:]

        # 生成摘要
        new_summary = self.summarize(messages_to_summarize, existing_summary)

        # 构建摘要消息
        summary_message = SystemMessage(content=f"[对话摘要]\n{new_summary}")

        # 返回：摘要 + 最近消息
        return [summary_message] + recent_messages, new_summary


# 全局摘要器
_default_summarizer: Optional[MessageSummarizer] = None


def get_summarizer(
    max_messages: int = 20,
    keep_recent: int = 4
) -> MessageSummarizer:
    """获取摘要器"""
    global _default_summarizer
    if _default_summarizer is None:
        _default_summarizer = MessageSummarizer(max_messages=max_messages, keep_recent=keep_recent)
    return _default_summarizer

3.3 集成到聊天服务

修改 services/langgraph_chat.py，添加裁剪/摘要支持：

"""在现有 langgraph_chat.py 中添加裁剪/摘要支持"""
import logging
from services.message_trimmer import get_trimmer
from services.message_summarizer import get_summarizer

logger = logging.getLogger(__name__)


class LangGraphChatService:
    # ... 现有代码 ...

    def __init__(self, config: Optional[ChatConfig] = None):
        self.config = config or ChatConfig()
        # 添加裁剪器和摘要器
        self.trimmer = get_trimmer(max_tokens=4000)
        self.summarizer = get_summarizer(max_messages=20, keep_recent=4)
        self.use_summarization = True  # 是否使用摘要（默认开启）

    def _call_model_with_trimming(
        self,
        state: MessagesState,
        model: Optional[str] = None,
        system_prompt: Optional[str] = None,
        conversation_summary: Optional[str] = None
    ):
        """调用模型（带消息裁剪/摘要）"""
        messages = state["messages"]

        # 统计原始消息
        stats = self.trimmer.get_stats(messages)
        logger.debug(f"原始消息: {stats}")

        # 根据策略处理消息
        if self.use_summarization and self.summarizer.should_summarize(messages):
            # 使用摘要策略
            processed_messages, new_summary = self.summarizer.process_with_summary(
                messages, conversation_summary
            )
            logger.info(f"使用摘要策略，消息数: {len(messages)} -> {len(processed_messages)}")
        else:
            # 使用裁剪策略
            processed_messages = self.trimmer.trim(messages)
            new_summary = conversation_summary

        # 添加系统提示
        if system_prompt:
            processed_messages = [
                SystemMessage(content=system_prompt)
            ] + processed_messages

        # 调用 LLM
        llm = self._get_llm(model)
        response = llm.invoke(processed_messages)

        return {
            "messages": [response],
            "summary": new_summary  # 可选：存储摘要到状态
        }

    # ... 修改 _build_graph 使用新方法 ...

---

四、API 扩展

4.1 添加配置端点

在 api/chat.py 中添加：

@router.get("/chat/config")
async def get_chat_config():
    """获取聊天配置"""
    return {
        "success": True,
        "config": {
            "max_tokens": 4000,
            "max_messages": 20,
            "summarization_enabled": True
        }
    }


@router.put("/chat/config")
async def update_chat_config(
    use_summarization: bool = Query(...),
    max_tokens: int = Query(4000),
    request: Request = None
):
    """更新聊天配置（管理员）"""
    # TODO: 实现配置更新
    chat_service = get_chat_service()
    chat_service.use_summarization = use_summarization
    chat_service.trimmer.max_tokens = max_tokens

    return {"success": True}

---

五、前端展示

5.1 添加摘要展示

当使用摘要时，前端可以显示摘要指示器：

// 在 static/js/chat.js 中添加
class ConversationSummary {
    constructor() {
        this.summary = null;
        this.container = null;
    }

    setSummary(summary) {
        this.summary = summary;
        this.render();
    }

    render() {
        if (!this.summary) return;

        if (!this.container) {
            this.container = document.createElement('div');
            this.container.className = 'conversation-summary';
            this.container.innerHTML = `
                <div class="summary-toggle" onclick="conversationSummary.toggle()">
                    📝 对话摘要
                </div>
                <div class="summary-content" style="display: none;">
                    ${this.summary}
                </div>
            `;

            // 插入到消息列表顶部
            const messagesContainer = document.getElementById('messages-container');
            if (messagesContainer) {
                messagesContainer.insertBefore(this.container, messagesContainer.firstChild);
            }
        } else {
            this.container.querySelector('.summary-content').textContent = this.summary;
        }
    }

    toggle() {
        const content = this.container.querySelector('.summary-content');
        content.style.display = content.style.display === 'none' ? 'block' : 'none';
    }
}

const conversationSummary = new ConversationSummary();

5.2 CSS 样式

/* 对话摘要样式 */
.conversation-summary {
    background: #f0f4f8;
    border-radius: 8px;
    margin: 10px 20px;
    overflow: hidden;
}

.summary-toggle {
    padding: 10px 15px;
    cursor: pointer;
    font-size: 14px;
    color: #666;
}

.summary-toggle:hover {
    background: #e4e8ec;
}

.summary-content {
    padding: 15px;
    font-size: 13px;
    color: #444;
    border-top: 1px solid #ddd;
    line-height: 1.6;
}

---

六、状态存储（可选）

6.1 在数据库中存储摘要

如果需要持久化摘要，可以在消息表中添加摘要字段：

# 在 models/message.py 中添加
class Message(Base):
    __tablename__ = "messages"

    id = Column(Integer, primary_key=True)
    conversation_id = Column(Integer, ForeignKey("conversations.id"))
    role = Column(String(20))
    content = Column(Text)
    # ... 其他字段 ...

    # 新增：摘要字段（仅用于特殊记录）
    is_summary = Column(Boolean, default=False)  # 标记是否为摘要记录

---

七、测试计划

7.1 单元测试

# tests/test_trimming.py
import pytest
from langchain_core.messages import HumanMessage, AIMessage
from services.message_trimmer import MessageTrimmer, count_tokens_approximately


def test_no_trimming_needed():
    """测试不需要裁剪的情况"""
    trimmer = MessageTrimmer(max_tokens=1000)
    messages = [
        HumanMessage(content="你好"),
        AIMessage(content="你好！有什么可以帮助你的？"),
    ]

    result = trimmer.trim(messages)
    assert len(result) == 2


def test_trimming_applied():
    """测试裁剪生效"""
    trimmer = MessageTrimmer(max_tokens=10)  # 很小的阈值

    # 创建长消息列表
    messages = []
    for i in range(20):
        messages.append(HumanMessage(content=f"这是第 {i} 条消息，内容比较长" * 10))
        messages.append(AIMessage(content=f"这是第 {i} 条回复，内容也比较长" * 10))

    result = trimmer.trim(messages)
    assert len(result) < len(messages)


def test_token_counter():
    """测试 token 计数"""
    msg = HumanMessage(content="你好世界")  # 4 个中文字符
    count = count_tokens_approximately(msg)
    assert count >= 1


# tests/test_summarization.py
def test_should_summarize():
    """测试摘要触发条件"""
    summarizer = MessageSummarizer(max_messages=10)

    # 少于阈值
    short_messages = [HumanMessage(content=f"消息 {i}") for i in range(5)]
    assert not summarizer.should_summarize(short_messages)

    # 超过阈值
    long_messages = [HumanMessage(content=f"消息 {i}") for i in range(15)]
    assert summarizer.should_summarize(long_messages)


def test_format_conversation():
    """测试对话格式化"""
    summarizer = MessageSummarizer()
    messages = [
        HumanMessage(content="你好"),
        AIMessage(content="你好！"),
    ]

    text = summarizer._format_conversation(messages)
    assert "用户" in text
    assert "AI" in text

---

八、实施步骤

步骤 1: 实现 Trimming（0.25 天）

• 创建 services/message_trimmer.py
• 实现 trim 方法
• 编写单元测试

步骤 2: 实现 Summarization（0.25 天）

• 创建 services/message_summarizer.py
• 实现 summarize 方法
• 编写单元测试

步骤 3: 集成到聊天服务（0.25 天）

• 修改 services/langgraph_chat.py
• 添加配置选项
• 测试集成

步骤 4: 前端展示（0.25 天）

• 添加摘要展示组件
• 添加 CSS 样式
• 测试用户体验

---

九、配置建议

9.1 不同场景的配置

场景	max_tokens	max_messages	策略
免费用户	2000	10	Trimming
标准用户	4000	20	Summarization
高级用户	8000	40	Summarization

9.2 模型选择

摘要生成建议使用较小的模型：

• openai/gpt-4o-mini
• anthropic/claude-3.5-haiku

这样可以降低成本，同时保证摘要质量。

---

十、相关文档

• LangGraph 模块使用分析
• LangGraph Memory 架构详解